package com.mini.grpc.server;

import com.mini.grpc.transport.netty.NettyServerTransport;
import com.mini.grpc.transport.ServerTransport;
import com.mini.grpc.transport.ServerListener;
import com.mini.grpc.transport.ServerStream;
import com.mini.grpc.transport.ServerStreamListener;
import com.mini.grpc.common.Status;
import com.mini.grpc.common.Metadata;
import com.mini.grpc.common.MethodDescriptor;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.logging.Logger;

/**
 * 服务端实现类
 * 
 * @author Mini-gRPC
 */
public class ServerImpl implements Server {
    
    private static final Logger logger = Logger.getLogger(ServerImpl.class.getName());
    
    private final int port;
    private final List<ServerServiceDefinition> services;
    private final List<ServerInterceptor> interceptors;
    private final Executor executor;
    private final long handshakeTimeoutMillis;
    private final int maxInboundMessageSize;
    private final int maxInboundMetadataSize;
    
    private final AtomicBoolean started = new AtomicBoolean(false);
    private final AtomicBoolean shutdown = new AtomicBoolean(false);
    private final AtomicBoolean terminated = new AtomicBoolean(false);
    private final CountDownLatch terminationLatch = new CountDownLatch(1);
    private final AtomicReference<ServerTransport> transportRef = new AtomicReference<>();
    
    private volatile int actualPort = -1;
    
    /**
     * 构造函数
     */
    public ServerImpl(int port,
                     List<ServerServiceDefinition> services,
                     List<ServerInterceptor> interceptors,
                     Executor executor,
                     long handshakeTimeoutMillis,
                     int maxInboundMessageSize,
                     int maxInboundMetadataSize) {
        this.port = port;
        this.services = Collections.unmodifiableList(new ArrayList<>(services));
        this.interceptors = Collections.unmodifiableList(new ArrayList<>(interceptors));
        this.executor = executor;
        this.handshakeTimeoutMillis = handshakeTimeoutMillis;
        this.maxInboundMessageSize = maxInboundMessageSize;
        this.maxInboundMetadataSize = maxInboundMetadataSize;
    }
    
    @Override
    public Server start() {
        if (!started.compareAndSet(false, true)) {
            throw new IllegalStateException("Server already started");
        }
        
        try {
            // 创建服务端监听器
            ServerListener listener = new ServerListenerImpl();
            
            // 创建服务端传输层
            ServerTransport transport = new NettyServerTransport(
                new java.net.InetSocketAddress(port), 
                listener,
                false // 暂时不使用SSL
            );
            
            // 启动传输层
            transport.start();
            transportRef.set(transport);
            
            // 获取实际端口
            this.actualPort = transport.getPort();
            
            logger.info("Server started on port " + actualPort);
            return this;
            
        } catch (Exception e) {
            started.set(false);
            throw new RuntimeException("Failed to start server", e);
        }
    }
    
    @Override
    public int getPort() {
        return actualPort;
    }
    
    @Override
    public List<ServerServiceDefinition> getServices() {
        return services;
    }
    
    @Override
    public List<ServerServiceDefinition> getImmutableServices() {
        return services; // 已经是不可变的
    }
    
    @Override
    public Server shutdown() {
        if (!shutdown.compareAndSet(false, true)) {
            return this;
        }
        
        ServerTransport transport = transportRef.get();
        if (transport != null) {
            transport.shutdown(Status.OK.withDescription("Server shutdown requested"));
        }
        
        logger.info("Server shutdown initiated");
        return this;
    }
    
    @Override
    public Server shutdownNow() {
        if (!shutdown.compareAndSet(false, true)) {
            return this;
        }
        
        ServerTransport transport = transportRef.get();
        if (transport != null) {
            transport.shutdown(Status.CANCELLED.withDescription("Server shutdown now requested"));
        }
        
        logger.info("Server shutdown now initiated");
        return this;
    }
    
    @Override
    public boolean isShutdown() {
        return shutdown.get();
    }
    
    @Override
    public boolean isTerminated() {
        return terminated.get();
    }
    
    @Override
    public void awaitTermination() throws InterruptedException {
        terminationLatch.await();
    }
    
    @Override
    public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException {
        return terminationLatch.await(timeout, unit);
    }
    
    /**
     * 服务端监听器实现
     */
    private class ServerListenerImpl implements ServerListener {
        
        @Override
        public <ReqT, RespT> ServerStreamListener<ReqT> streamCreated(
                ServerStream<RespT> stream,
                MethodDescriptor<ReqT, RespT> method,
                Metadata headers,
                java.net.SocketAddress remoteAddress) {
            logger.info("New stream created: " + method.getFullMethodName());
            
            // 查找对应的服务处理器
            ServerCallHandler<ReqT, RespT> handler = findHandler(method);
            if (handler == null) {
                logger.warning("No handler found for method: " + method.getFullMethodName());
                // 返回一个错误处理器
                return new ServerStreamListener<ReqT>() {
                    @Override
                    public void messageReceived(ReqT message) {
                        stream.close(Status.UNIMPLEMENTED.withDescription("Method not found: " + method.getFullMethodName()), new Metadata());
                    }
                    
                    @Override
                    public void halfClosed() {
                        stream.close(Status.UNIMPLEMENTED.withDescription("Method not found: " + method.getFullMethodName()), new Metadata());
                    }
                    
                    @Override
                    public void cancelled(Status status) {
                        logger.info("Stream cancelled: " + status);
                    }
                };
            }
            
            // 创建ServerCall
            ServerCall<ReqT, RespT> call = new ServerCallImpl<>(stream, method);
            
            // 启动调用处理
            ServerCall.Listener<ReqT> callListener = handler.startCall(call, headers);
            
            // 返回适配器，将ServerStreamListener事件转换为ServerCall.Listener事件
            return new ServerStreamListener<ReqT>() {
                @Override
                public void messageReceived(ReqT message) {
                    logger.info("Message received: " + message);
                    callListener.onMessage(message);
                }
                
                @Override
                public void halfClosed() {
                    logger.info("=== Stream half closed, calling onHalfClose ===");
                    callListener.onHalfClose();
                }
                
                @Override
                public void cancelled(Status status) {
                    logger.info("Stream cancelled: " + status);
                    callListener.onCancel();
                }
            };
        }
        
        @Override
        public void transportTerminated() {
            terminated.set(true);
            terminationLatch.countDown();
            logger.info("Server transport terminated");
        }
    }
    
    /**
     * 查找方法对应的处理器
     */
    @SuppressWarnings("unchecked")
    private <ReqT, RespT> ServerCallHandler<ReqT, RespT> findHandler(MethodDescriptor<ReqT, RespT> method) {
        String fullMethodName = method.getFullMethodName();
        logger.info("=== Looking for handler for method: " + fullMethodName + " ===");
        logger.info("Available services: " + services.size());
        
        for (ServerServiceDefinition service : services) {
            logger.info("Checking service: " + service.getServiceDescriptor().getName());
            ServerMethodDefinition<?, ?> methodDef = service.getMethod(fullMethodName);
            if (methodDef != null) {
                logger.info("Found handler for method: " + fullMethodName);
                return (ServerCallHandler<ReqT, RespT>) methodDef.getServerCallHandler();
            }
        }
        
        logger.warning("No handler found for method: " + fullMethodName);
        return null;
    }
} 